package com.qf.kafka.spring.boot.demo;

import cn.hutool.json.JSONUtil;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.*;
import java.util.concurrent.ExecutionException;

public class MySimpleProducer {

    private final static String TOPIC_NAME = "my-replicated-topic";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1.设置参数
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "1");
            /*
            发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造
            成消息重复发送，⽐如⽹络抖动，所以需要在
            接收者那边做好消息接收的幂等性处理
            */
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        //重试间隔设置
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);


        //2.创建⽣产消息的客户端，传⼊参数
        Producer<String, List<Map<String, String>>> producer = new KafkaProducer<>(props);
        //3.创建消息
        //key：作⽤是决定了往哪个分区上发，value：具体要发送的消息内容
        List<Map<String, String>> records = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Map<String, String> record = new HashMap<>();
            record.put("ip", "1.1.1." + i);
            records.add(record);
        }
//        String jsonStr = JSONUtil.toJsonStr(records);
        ProducerRecord<String, List<Map<String, String>>> producerRecord = new ProducerRecord<>(TOPIC_NAME, "xzq", records);
        //4.发送消息,得到消息发送的元数据并输出
        RecordMetadata metadata = producer.send(producerRecord).get();
        System.out.println("同步⽅式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
    }





}


